package mq.consumer;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;

/**
 * @Author:sjb
 * @CreateTime:2024-08-20
 * @Description: 消费者类
 * @Version：1.0
 */
@Component
public class ConsumerExample {
    @Value("${training.mq.endPoint}")
    private static String endPoint;

    @Value("${training.mq.consumerGroup}")
    private static String consumerGroup;

    private static final Logger logger = LoggerFactory.getLogger(ConsumerExample.class);

    public static void main(String[] args) throws IOException {
        final ClientServiceProvider clientServiceProvider = ClientServiceProvider.loadService();
        ClientConfiguration configuration = ClientConfiguration.newBuilder().setEndpoints(endPoint).build();
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        PushConsumer pushConsumer = null;
        try {
             pushConsumer = clientServiceProvider.newPushConsumerBuilder()
                    .setConsumerGroup(consumerGroup)
                    .setClientConfiguration(configuration)
                    .setSubscriptionExpressions(Collections.singletonMap(tag,filterExpression))
                    .setMessageListener(messageView -> {
                        // todo 处理消息并返回消费结果
                        logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                        return ConsumeResult.SUCCESS;
                    }).build();
            Thread.sleep(1000);
        } catch (ClientException | InterruptedException e) {
            logger.error("Consume message unSuccessfully, messageId={}", e.getMessage());
        }finally {
            pushConsumer.close();
        }
    }
}
